package com.atguigu.tingshu.album.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.IdUtil;
import com.atguigu.tingshu.album.mapper.AlbumAttributeValueMapper;
import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.mapper.TrackStatMapper;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.common.cache.GuiGuCache;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.*;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.Nullable;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
@SuppressWarnings({"all"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

	@Autowired
	private AlbumInfoMapper albumInfoMapper;

	@Autowired
	private AlbumStatMapper albumStatMapper;

	@Autowired
	private AlbumAttributeValueMapper albumAttributeValueMapper;

	@Autowired
	private KafkaService kafkaService;

	@Autowired
	private RedisTemplate redisTemplate;

	@Autowired
	private TrackStatMapper trackStatMapper;

	@Autowired
	private StringRedisTemplate stringRedisTemplate;

	@Autowired
	private RedissonClient redissonClient;

	@Override
	public Page<AlbumListVo> getUserAlbumPage(Page<AlbumListVo> pageInfo, AlbumInfoQuery albumInfoQuery) {
		return albumInfoMapper.getUserAlbumPage(pageInfo, albumInfoQuery);
	}


	@Override
	@Transactional(rollbackFor = Exception.class)
	public void removeAlbumInfo(Long id) {

		AlbumInfo albumInfo = albumInfoMapper.selectById(id);
		if(albumInfo == null || albumInfo.getIncludeTrackCount() > 0){
			throw new GuiguException(400,"当前专辑下存在音频");
		}

		// 删除专辑
		albumInfoMapper.deleteById(id);


		LambdaQueryWrapper<AlbumStat> albumStatLambdaQueryWrapper = new LambdaQueryWrapper<>();
		albumStatLambdaQueryWrapper.eq(AlbumStat::getAlbumId,id);
		albumStatMapper.delete(albumStatLambdaQueryWrapper);

		//3.根据专辑ID删除专辑标签
		LambdaQueryWrapper<AlbumAttributeValue> albumAttributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
		albumAttributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
		albumAttributeValueMapper.delete(albumAttributeValueLambdaQueryWrapper);

		// kafka发送消息
		kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, albumInfo.getId().toString());
	}


	/**
	 * 利用 redisTemplate 缓存
	 * @param id
	 * @return
	 */
	@Override
	@GuiGuCache(prefix = RedisConstant.ALBUM_INFO_PREFIX)
	public AlbumInfo getAlbumInfo(Long id) {

		// 使用 redisson 实现
//		return getRedissonAlbumInfo(id);
		return this.getAlbumInfoDb(id);
		// 使用 redisTemplate 实现
	//	return getRedisAlbumInfo(id);
	}


	@Nullable
	private AlbumInfo getRedissonAlbumInfo(Long id){

		try {
			// 从redis中查
			String key = RedisConstant.ALBUM_INFO_PREFIX + id;
			AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(key);
			// 不为空直接返回
			if(albumInfo != null) {
				log.info("命中缓存，直接返回，线程ID：{}，线程名称：{}", Thread.currentThread().getId(), Thread.currentThread().getName());
				return albumInfo;
			}
			// 否者获取分布式锁,只让一个线程去DB查询
			String lockKey = RedisConstant.ALBUM_LOCK_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;

			// ------------------------开始自旋---------------------------//
			RLock lock = redissonClient.getLock(lockKey);
			lock.lock();  // 默认的锁会有看门狗机制,所以我们无需手动设置时间

			try {
				// 再次从redis中查
				albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(key);
				if (albumInfo != null){
					log.info("当前线程{},获取锁成功，且再次命中缓存成功", Thread.currentThread().getName());
					return albumInfo;
				}else{
					// 否者从DB中获取
					albumInfo = this.getAlbumInfoDb(id);

					if(albumInfo == null) albumInfo = new AlbumInfo();

					// 防止缓存穿透
					long ttl = albumInfo == null ? RedisConstant.ALBUM_TEMPORARY_TIMEOUT : RedisConstant.ALBUM_TIMEOUT;
					// 放入缓存
					redisTemplate.opsForValue().set(key,albumInfo,ttl,TimeUnit.SECONDS);
					return albumInfo;
				}
			}finally {
				log.info("当前线程：{}，释放锁", Thread.currentThread().getName());
				lock.unlock();
			}
		}catch (Exception e){
			log.error("[专辑服务]查询专辑数据异常：{}", e);
			return getAlbumInfoDb(id);
		}
	}


	@Nullable
	private AlbumInfo getRedisAlbumInfo(Long id) {
		try {
			// 首先到redis去查
			String key = RedisConstant.ALBUM_INFO_PREFIX + id;
			AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(key);

			// 如果存在直接返回，不存在再去数据库中查，利用分布式锁只放一个线程去查DB,其他线程自旋
			if(albumInfo != null) {
				log.info("命中缓存，直接返回，线程ID：{}，线程名称：{}", Thread.currentThread().getId(), Thread.currentThread().getName());
				return albumInfo;
			}

			// 获取分布式锁
			String lock_key = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
			String lock_value = IdUtil.fastSimpleUUID();  // 使用UUID 防止误删锁，只有key和value匹配上了才可以删
			Boolean lock = redisTemplate.opsForValue().setIfAbsent(lock_key, lock_value, RedisConstant.ALBUM_LOCK_EXPIRE_PX2, TimeUnit.SECONDS);

			if (lock){
				// 获取到了锁
				try {
					// 查询DB
					albumInfo = this.getAlbumInfoDb(id);
					if(albumInfo == null) albumInfo = new AlbumInfo();

					// 判断是否命中，没命中就存入一个null值，防止缓存穿透
					long ttl = albumInfo == null ? RedisConstant.ALBUM_TEMPORARY_TIMEOUT : RedisConstant.ALBUM_TIMEOUT;
					redisTemplate.opsForValue().set(key,albumInfo,ttl,TimeUnit.SECONDS);
					return albumInfo;
				}finally {
					// 业务完成释放锁
					String scriptText = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
							"then\n" +
							"    return redis.call(\"del\",KEYS[1])\n" +
							"else\n" +
							"    return 0\n" +
							"end";
					DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
					redisScript.setScriptText(scriptText);
					redisScript.setResultType(Long.class);
					redisTemplate.execute(redisScript, Arrays.asList(lock_key), lock_value);
				}

			}else{
				// 没有获取到锁,那就让线程自旋
				try {
					Thread.sleep(200);
				} catch (InterruptedException e) {
					throw new RuntimeException(e);
				}
				log.error("获取锁失败，自旋：{}，线程名称：{}", Thread.currentThread().getId(), Thread.currentThread().getName());
				return getAlbumInfo(id);
			}
		}catch (Exception e){
			//兜底处理方案：Redis服务有问题，将业务数据获取自动从数据库获取
			log.info("[专辑服务]Redis服务异常：{}", e);
			return this.getAlbumInfoDb(id);
		}
	}

	@Nullable
	private AlbumInfo getAlbumInfoDb(Long id) {
		AlbumInfo albumInfo = albumInfoMapper.selectById(id);
		if(albumInfo != null){
			LambdaQueryWrapper<AlbumAttributeValue> albumAttributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
			albumAttributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId,albumInfo.getId());

			List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(albumAttributeValueLambdaQueryWrapper);
			albumInfo.setAlbumAttributeValueVoList(albumAttributeValueList);
		}
		return albumInfo;
	}

	@Override
	@Transactional(rollbackFor = Exception.class)
	public void updateAlbumInfo(Long id, AlbumInfoVo albumInfoVo) {
		//1.修改专辑信息
		//2.1 将修改专辑信息转为PO对象 封装专辑ID
		AlbumInfo albumInfo = BeanUtil.copyProperties(albumInfoVo, AlbumInfo.class);
		albumInfo.setId(id);
		//2.2 修改专辑
		albumInfoMapper.updateById(albumInfo);

		//2.修改专辑标签 先删再新增
		//2.1 根据专辑ID将以前关联标签记录全部删除
		LambdaQueryWrapper<AlbumAttributeValue> queryWrapper = new LambdaQueryWrapper<>();
		queryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
		albumAttributeValueMapper.delete(queryWrapper);

		//2.2 在根据提交专辑标签做新增
		List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
		if (CollectionUtil.isNotEmpty(albumAttributeValueVoList)) {
			//2.2 关联专辑ID批量保存专辑标签关系
			for (AlbumAttributeValueVo albumAttributeValueVo : albumAttributeValueVoList) {
				AlbumAttributeValue albumAttributeValue = BeanUtil.copyProperties(albumAttributeValueVo, AlbumAttributeValue.class);
				albumAttributeValue.setAlbumId(id);
				albumAttributeValueMapper.insert(albumAttributeValue);
			}
		}


		// 判断 是否设置私密
		if ("1".equals(albumInfo.getIsOpen())) {
			kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, albumInfo.getId().toString());
		} else {
			kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, albumInfo.getId().toString());
		}

	}

	@Override
	public List<AlbumInfo> findUserAllAlbumList(Long userId) {
		LambdaQueryWrapper<AlbumInfo> albumInfoLambdaQueryWrapper = new LambdaQueryWrapper<>();
		albumInfoLambdaQueryWrapper.eq(AlbumInfo::getUserId,userId);
		albumInfoLambdaQueryWrapper.select(AlbumInfo::getId,AlbumInfo::getAlbumTitle,AlbumInfo::getStatus);
		albumInfoLambdaQueryWrapper.last("limit 200");
		albumInfoLambdaQueryWrapper.orderByDesc(AlbumInfo::getId);
		return albumInfoMapper.selectList(albumInfoLambdaQueryWrapper);
	}

	@Override
	@GuiGuCache(prefix = "albumStatVo:")
	public AlbumStatVo getAlbumStatVo(Long albumId) {
		return albumStatMapper.getAlbumStatVo(albumId);
	}

	/**
	 *  MQ监听更新声音统计信息
	 * @param trackStatMqVo
	 */
	@Override
	@Transactional(rollbackFor = Exception.class)
	public void updateTrackStat(TrackStatMqVo mqVo) {
		//1.做幂等性处理，统一个消息只处理一次 采用set k(业务消息唯一标识) v NX EX
		String key = "mq:" + mqVo.getBusinessNo();
		try {
			Boolean ifAbsent = redisTemplate.opsForValue().setIfAbsent(key, mqVo.getBusinessNo(), 1, TimeUnit.HOURS);
			String statType = mqVo.getStatType();
			if (!ifAbsent) return;
			// 更新声音统计信息
			trackStatMapper.updateStat(mqVo.getTrackId(), mqVo.getStatType(), mqVo.getCount());
			// 更新专辑信息
			//3.更新专辑统计信息(播放量、评论量只要声音+1，对应专辑也得+1)
			if (SystemConstant.TRACK_STAT_PLAY.equals(mqVo.getStatType())) {
				albumStatMapper.updateStat(mqVo.getAlbumId(), SystemConstant.ALBUM_STAT_PLAY, mqVo.getCount());
			}
			if (SystemConstant.TRACK_STAT_COMMENT.equals(mqVo.getStatType())) {
				albumStatMapper.updateStat(mqVo.getAlbumId(), SystemConstant.ALBUM_STAT_COMMENT, mqVo.getCount());
			}
		}catch (Exception e){
			//如果更新数据库发送异常，事务会进行回滚，下次再次投递消息允许继续处理统一个消息
			redisTemplate.delete(key);
			throw new RuntimeException(e);
		}
	}

	/**
	 * 分布式锁
	 */
	@Override
	public void testLock() {
		//0.1 创建锁对象
		RLock lock = redissonClient.getLock("lock");

		//0.2 获取锁
		//0.2.1 一直等待到获取锁,如果获取锁成功,锁的有效时间为:30s,底层启动"看门狗"线程(如果业务有超时风险)可以延迟锁有效时间
		lock.lock();

		//0.2.2 一直等待到获取锁,如果获取锁成功,自定义锁有效时间
		//lock.lock(10, TimeUnit.SECONDS);

		//0.2.3 尝试获取锁 参数1:等待获取锁时间,超时则放弃获取  参数2:如果获取锁成功,锁的有效时间 参数3:时间单位
		//boolean b = lock.tryLock(3, 10, TimeUnit.SECONDS);
		try {
			//1.从Redis缓存中获取key="num"的值  保证redis中存在"num"(手动提前在redis中创建key)
			String value = stringRedisTemplate.opsForValue().get("num");
			if (StringUtils.isBlank(value)) {
				return;
			}
			//2.对获取到值进行+1操作
			int num = Integer.parseInt(value);
			stringRedisTemplate.opsForValue().set("num", String.valueOf(++num));

		} catch (NumberFormatException e) {
			e.printStackTrace();
		} finally {
			//3. 释放锁
			lock.unlock();
		}
	}

}


















